package com.lagou.spark
/*
3.1、用Spark-Core实现统计每个adid的曝光数与点击数，将结果输出到hdfs文件；
输出文件结构为adid、曝光数、点击数。注意：数据不能有丢失（存在某些adid有imp，没有clk；或有clk没有imp）
3.2、你的代码有多少个shuffle，是否能减少？
（提示：仅有1次shuffle是最优的）
 */
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object LogClick {
  def main(args: Array[String]): Unit = {
    //创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //读取文件
    val click: RDD[String] = sc.textFile("data/click.log")
    val imp: RDD[String] = sc.textFile("data/imp.log")

    //切分文件
    val clickRDD: RDD[(String, (Int, Int))] = click.map(x => (x.split("[&]")(4), (1, 0)))
    val impRDD: RDD[(String, (Int, Int))] = imp.map(x => (x.split("[&]")(4), (0, 1)))

    //聚合计算，求并集
    val result: RDD[(String, (Int, Int))] = clickRDD.union(impRDD)
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
    result.collect().foreach(println)
    
	//存入hdfs
    result.saveAsTextFile("hdfs://server1:9000/click_output/")
    
	//关闭SparkContext
    sc.stop()
  }

}
